package act;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executor;
import java.util.concurrent.SynchronousQueue;

/*********************************************************************
 * The ActorFactory should only be used to create a top level actor in a hierarchy. From there on in that actor should be used to create other actors.
 * 
 * @author poconnor
 * @version 1.0.0
 *********************************************************************/
public class ActorFactory {

	public static final int DEFAULT_MAIL_BOX_SIZE = 100;

	private static class Actor implements Runnable, IActor {
		private BlockingQueue<Object> _mailbox = null;
		private final int _mailboxSize;
		private final IReceiver _handler;
		private final ActorFactory _factory;
		private volatile boolean _stop = false;
		private volatile boolean _started = false;
		private final CountDownLatch _latchFinished = new CountDownLatch(1);

		/**
		 * @param mailboxSize
		 *            0 => unbuffered mailbox
		 */
		public Actor(ActorFactory factory, IReceiver handler, int mailboxSize) {
			_factory = factory;
			_handler = handler;
			_mailboxSize = mailboxSize;
			handler.setActor(this);
		}

		public Actor(ActorFactory factory, IReceiver handler, BlockingQueue<Object> mailbox) {
			_factory = factory;
			_handler = handler;
			_mailbox = mailbox;
			_mailboxSize = 0;
			handler.setActor(this);
		}

		@Override
		public void send(Object msg) {

			// start the message processing loop if its not started already
			start();

			try {
				// put a message on the queue.
				// These messages will be processed one at a time by the run() method of this actor.
				_mailbox.put(msg);

				// should we put some warning here if we receive a message after this actor has been stopped?
				// probably better not to. Its up to the user to know what they are doing
			} catch (InterruptedException e) {
				throw new RuntimeException(e);
			}

		}

		@Override
		public void run() {
			// Take a the next message off the queue and send it on to the handler
			// If we have been sent a stop signal then just stop this thread.

			while (!_stop) {
				Object msg;
				try {
					msg = _mailbox.take();
				} catch (InterruptedException e) {
					throw new RuntimeException(e);
				}
				// pass the message on to the handler. It will only ever receive one message at a time
				if (!_stop) {
					_handler.receive(msg);
				}
			}
			_latchFinished.countDown();// unset the latch so that any waiters can finish see this.await()
		}

		// Create the mailbox/queue and start the message processing loop if its not started already
		private void start() {
			if (!_started) {// do quick check first so that we don't have to have go into a synchronized block for every message we process
				synchronized (this) {
					// check started again in case another thread set it since the if(!started) check
					if (!_started) {
						// this is the first message that has been sent to this actor
						// we need to create the mailbox and kick off the thread that will run it
						if (_mailbox == null) {
							// the client didn't supply mailbox to the ctr so create one
							if (_mailboxSize <= 0) {
								// a queue of no size, its just a handoff point
								_mailbox = new SynchronousQueue<Object>();
							} else {
								_mailbox = new ArrayBlockingQueue<Object>(this._mailboxSize);
							}
							_factory.getExecutor().execute(this);
						}
						_started = true;
					}
				}
			}
		}

		@Override
		public void stop() {
			_stop = true;

			// Have to put one dummy message on the mailbox in because its likely
			// that it is now empty and that is why someone has called stop.
			// If the mailbox is empty the run method will be blocked on mailbox.take()
			// So we need to wake up the run method with this dummy message so that it
			// can then terminate normally.
			try {
				_mailbox.put(new Object());
			} catch (InterruptedException e) {
				throw new RuntimeException(e);
			}
		}

		@Override
		public void await() {
			try {
				_latchFinished.await();
			} catch (InterruptedException e) {
				throw new RuntimeException(e);
			}
		}

		@Override
		public IActor create(IReceiver handler) {

			return _factory.create(handler);
		}

		@Override
		public IActor create(IReceiver handler, BlockingQueue<Object> mailbox) {
			return _factory.create(handler, mailbox);
		}

		@Override
		public IActor create(IReceiver handler, int mailboxSize) {
			return _factory.create(handler, mailboxSize);
		}

	}

	private final Executor _executor;

	public ActorFactory(Executor executor) {
		_executor = executor;
	}

	Executor getExecutor() {
		return _executor;
	}

	/**
	 * @param handler
	 *            A handler to handle messages sent to the actor
	 * @param mailboxSize
	 *            If 0 is specified the mailbox will be unbuffered (specifically it will use a {@link java.util.concurrent.SynchronousQueue} The mailbox will only be created when the first message is sent to the actor.
	 */
	public IActor create(IReceiver handler, int mailboxSize) {

		// just wrap the handler in the actor
		// All the actor does is serialise any calls to the handler via a queue.
		Actor actor = new Actor(this, handler, mailboxSize);
		return actor;
	}

	/**
	 * Create a new actor with a handler with an mailbox of size 0 (SynchronousQueue)
	 */
	public IActor create(IReceiver handler) {
		return create(handler, 0);
	}

	/**
	 * Create a new actor with a handler with a specific implementation of BlockingQueue for its mailbox
	 * 
	 * @param _mailboxSize
	 *            If 0 is specified the mailbox will be unbuffered (specifically it will use a {@link java.util.concurrent.SynchronousQueue}
	 */
	public IActor create(IReceiver handler, BlockingQueue<Object> mailbox) {

		// just wrap the handler in the actor
		// All the actor does is serialise any calls to the handler via a queue.
		Actor actor = new Actor(this, handler, mailbox);
		return actor;
	}

}
